package com.qisi.demo;

import com.qisi.functions.LogBeanFlatMapFunction;
import com.qisi.functions.SourceFunctions;
import com.qisi.pojo.SentryLog;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KafkaDemo {
    public static void main(String[] args) throws Exception {
        // 全局配置
        Configuration config = ParameterTool.fromArgs(args).getConfiguration();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取 Kafka 日志流
        DataStreamSource<String> kafkaSource = env.addSource(SourceFunctions.getKafkaEventSource());

        // 筛选出 MessageLog
        // SingleOutputStreamOperator<SentryLog.SentryData> msg = kafkaSource.flatMap(new LogBeanFlatMapFunction());

        kafkaSource.print();

        env.execute("KafkaDemo");

    }
}
